/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;

/**
 * Base class for activities that read one or more elements from input, compute and
 * store result, and then, output zero or more elements when output channel is available.
 * 
 * @author Alex Antonau
 * */
public abstract class StatefulActivity<I,O> extends AbstractActivity<I, O> {

	private State state=State.READY;
		
	public State getState(ActivityContext<I, O> ctx) {
		return state;
	}

	public void setState(ActivityContext<I, O> ctx, State state) {
		this.state = state;
	}

	public boolean isReady(ActivityContext<I, O> ctx) throws Exception {
		State state = getState(ctx);
		switch (state) {
			case WAIT: return false;
			case PROCESS: return true;
			case OUTPUT: return ctx.getOutput().freeSize() >= getOutputSize(ctx);
		}
		if(ctx.getOutput().isClosed()) return false;
		final Channel<I> input = ctx.getInput();
		final int available = input.size();
		if(available == 0) return false;
		return available >= getInputSize(ctx) || input.isClosed();
	}

	public boolean isDone(ActivityContext<I, O> ctx) throws Exception {
		boolean done = false;
		final Channel<I> input = ctx.getInput();
		final Channel<O> output = ctx.getOutput();
		State state = getState(ctx);
		if(state==State.READY && input.isClosed() && input.isEmpty()) done=true;
		if(done){
			if(input.hasFailed()) output.fail(input.getFailureCause());
		}else{
			if(output.isClosed()) done = true;
			if(done){
				if(output.hasFailed()) input.fail(output.getFailureCause());
				else {
					input.close();
				}
			}
		}
		return done;
	}
	public void execute(ActivityContext<I, O> ctx) throws Exception {
		State state = getState(ctx);
		if(state == State.READY){
			final int inputSize = getInputSize(ctx);
			final Channel<I> input = ctx.getInput();
			Lock<I> readLock = input.getReadLock(inputSize);
			if(readLock==null){
				final int available = input.size();
				if(input.isClosed() && available >0 && available < inputSize) 
					readLock = input.getReadLock(available);
			}
			if(readLock!=null){
				try{
					input(ctx, readLock);
				} finally{
					readLock.unlock();
				}
				setState(ctx, State.PROCESS);
			}else return;
			state = getState(ctx);
		}
		if(state == State.PROCESS){
			try{
				process(ctx);
			} finally{
				setState(ctx, State.OUTPUT);
			}
			state = getState(ctx);
		}
		if(state==State.OUTPUT){
			final int outputSize = getOutputSize(ctx); 
			if(outputSize > 0){
				Lock<O> writeLock = ctx.getWriteLock(outputSize); 
				if(writeLock!=null){
					try{
						output(ctx, writeLock);
					} finally{
						writeLock.unlock();
						setState(ctx, State.READY);
					}
				}
			}else setState(ctx, State.READY);
		}
	}
	
	/**
	 * Number of output elements required for result.
	 * @return integer that equal or greater than 0.
	 * */
	public int getOutputSize(ActivityContext<I, O> ctx){
		return 1;
	}
	
	/**
	 * Number of input elements required for computation.
	 * @return integer that equal or greater than 1.
	 * */
	public int getInputSize(ActivityContext<I, O> ctx){
		return 1;
	}
	/**
	 * Called if state is READY and getInputSize() number of elements is available in the input channel.
	 * Implement this method to read input data into memory. 
	 * */
	public abstract void input(ActivityContext<I, O> ctx, Lock<I> inputLock)throws Exception;
	/**
	 * A method that implements the computational work load. 
	 * During method execution activity does not hold any locks. 
	 * */
	public void process(ActivityContext<I, O> ctx) throws Exception {}
	/**
	 * Called if state is OUTPUT and getOutputSize() returns a number > 0.
	 * Lock size may be lesser than requested output size. In this case output may 
	 * be called several times. Lock size should be subtracted from the number 
	 * returned by getOutputSize.  
	 * */
	public abstract void output(ActivityContext<I, O> ctx, Lock<O> outputLock)throws Exception;

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append(getClass().getName()).append("[state=");
		sb.append(state).append("]");
		return sb.toString();
	}
	
}
